<?php


namespace App;


use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class Producer
{
    /**
     * @var AMQPStreamConnection
     */
    private $rabbitmq;
    /**
     * @var AMQPChannel
     */
    private $channel;

    public function __construct(AMQPStreamConnection $connection)
    {
        $this->rabbitmq = $connection;
        $this->channel = $this->rabbitmq->channel();
    }

    /**
     * 简单，工作模式
     */
    public function helloWorld()
    {
        $queueName = 'hello';
        // 声明队列，如果有就直接用，没有则创建
        // 第三个参数表示消息是否持久化
        $this->channel->queue_declare($queueName, false, true, false, false);

        $number = 1;
        while ($number <= 20) {
            $msgStr = "[$number]:". 'Hello World!';
            // 创建消息对象
            $msg = new AMQPMessage( $msgStr);
            // 发布消息
            $this->channel->basic_publish($msg, '', $queueName);
            echo '[发布消息]: ' . $msgStr . "\n";
            $number++;
        }
        // 关闭管道
        $this->channel->close();
    }

    /**
     * 发布/订阅模式
     */
    public function publishSubscribe()
    {
        // 声明交换机
        $this->channel->exchange_declare('logs', 'fanout', false, false, false);

        $number = 1;
        while (1 || $number <= 20) {
            $msgStr = "[$number]:". rand(10000000, 99999999) . 'log log log log log!';
            // 创建消息对象
            $msg = new AMQPMessage($msgStr);
            // 发布消息
            $this->channel->basic_publish($msg, 'logs');
            echo '[发布消息]: ' . $msgStr . "\n";
            $number++;

        }
        $this->channel->close();
    }

    /**
     * 路由模式
     */
    public function routing()
    {
        // 声明交换机
        $this->channel->exchange_declare('direct_log', 'direct', false, false, false);

        $number = 1;
        $severity = ['info', 'debug', 'error'];
        while (1 || $number <= 20) {
            $routing = rand(0, 2);
            $msgStr = "[$number]:". $severity[$routing] . ' log log log log log!';
            // 创建消息对象
            $msg = new AMQPMessage($msgStr);
            // 发布消息
            $this->channel->basic_publish($msg, 'direct_log', $severity[$routing]);
            echo '[发布消息]: ' . $msgStr . "\n";
            $number++;

        }
        $this->channel->close();
    }

    /**
     * 主题模式
     */
    public function topic()
    {
        // 声明交换机
        $this->channel->exchange_declare('topic_logs', 'topic', false, false, false);
        $number = 1;
        $severity = [
            'quick.orange.rabbit',
            'lazy.orange.elephant',
            'quick.orange.fox',
            'lazy.brown.fox',
            'lazy.pink.rabbit',
            'quick.brown.fox',
            'quick.orange.male.rabbit',
            'lazy.orange.male.rabbit'
        ];
        while ($number <= 20) {
            $routing = rand(0, 7);
            $msgStr = "[$number]:". $severity[$routing] . ' log log log log log!';
            // 创建消息对象
            $msg = new AMQPMessage($msgStr);
            // 发布消息
            $this->channel->basic_publish($msg, 'topic_logs', $severity[$routing]);
            echo '[发布消息]: ' . $msgStr . "\n";
            $number++;

        }
        $this->channel->close();
    }

    /**
     * rpc
     */
    public function rpc()
    {
        echo "[x] 等待rpc调用\n";
        $this->channel->queue_declare('rpc_queue', false, false, false, false);
        $this->channel->basic_qos(null, 1, null);
        $this->channel->basic_consume('rpc_queue', '', false, false, false, false, function (AMQPMessage $req) {
            $n = intval($req->body);
            echo "[x] 调用 fib($n);\n";
            $msg = new AMQPMessage((string)static::fib($n), [
                'correlation_id' => $req->get('correlation_id')
            ]);
            /**
             * @var AMQPChannel $channel
             */
            $channel = $req->delivery_info['channel'];
            $channel->basic_publish($msg, '', $req->get('reply_to'));
            $req->ack();
        });

        while ($this->channel->is_consuming()) {
            $this->channel->wait();
        }
        $this->channel->close();
    }

    private static function fib($n)
    {
        if ($n == 0) {
            return 0;
        }
        if ($n == 1) {
            return 1;
        }
        return static::fib($n - 1) + static::fib($n - 2);
    }
}